
#include "common.h"
#include "thread.h"
#include "globals.h"
#include "http.h"
#include "server.h"

static int init_count = 0;
static pthread_mutex_t init_lock;
static pthread_cond_t init_cond;


static void *thread_loop(void *arg);
static void thread_event_setup(thread_t *t);
static void worker_thread_create(void* (*handler)(void *), void *arg);
static void thread_libevent_handle(int fd, short flag, void *arg);


thread_t *thread_select(void)
{
	int min ;
	int pos ;
	int i;
	
	min = queue_length(threads[1].active_conns);
	pos = 1;
	for(i = 1; i <= nworker; i++) {
		int len = queue_length(threads[i].active_conns);
		if(len < min) {
			min = len;
			pos = i;	
		}
	}
	
	debug_log(3, "select thread %d, id %u\n", pos, threads[pos].id);
	return &threads[pos];
}

void thread_init(int number)
{
	int factual = 2;
	int i;
	
	pthread_mutex_init(&init_lock, NULL);
	pthread_cond_init(&init_cond, NULL);
	
	if(number > 0) 
		factual = number;
	
	nworker = factual - 1;

	threads = mempool_alloc(factual * sizeof(*threads));
	if(!threads) {
		error_log("init threads %d failed\n", factual);
		abort();
	}

	threads[0].base = master_base;
	threads[0].id = pthread_self();


	for (i = 0; i < factual; i++) {
		int fds[2];
		if (pipe(fds)) {
			error_log("Can't create notify pipe. %d\n", i);
			exit(1);
		}

		threads[i].notify_receive_fd = fds[0];
		threads[i].notify_send_fd = fds[1];
		/* init conns queue */
		threads[i].waiting_conns = (queue_t *)queue_create();
		threads[i].active_conns = (queue_t *)queue_create();

		thread_event_setup(&threads[i]);
	}

	/* create thread */
	for(i = 1; i < factual; i++)
		worker_thread_create(thread_loop, &threads[i]);

	/* waiting for all children threads */
	pthread_mutex_lock(&init_lock);
	init_count++; /* main thread */
	while (init_count < factual) {
		pthread_cond_wait(&init_cond, &init_lock);
	}
	pthread_mutex_unlock(&init_lock);

	debug_log(1, "1 master, %d worker running ...\n", nworker);
}

void thread_clean(void)
{
	int i;
	thread_t *cur = NULL;
	struct list_head *p, *n;

	debug_log(5, "discard %d threads resource\n", conf->server.worker);
	/* break all active connections */
	for(i = 0; i < conf->server.worker; i++) {
		cur = &threads[i];
		if(!cur)
			continue;
		event_base_loopbreak(cur->base);
		/* free all waiting queue and active queue */
		list_for_each_safe(p, n, &cur->waiting_conns->list) {
			accept_t *at = list_entry(p, accept_t, list);
			list_del(p);
			mempool_free(at);
		}
		queue_destroy(cur->waiting_conns);
		list_for_each_safe(p, n, &cur->active_conns->list) {
			conn_t *con = list_entry(p, conn_t, list);
			list_del(p);
			conn_free(con);
		}
		queue_destroy(cur->active_conns);
		close(cur->notify_send_fd);
		close(cur->notify_receive_fd);
	}
	
	mempool_free(threads);
	threads = NULL;
}

static void thread_event_setup(thread_t *me)
{
	if(!me) 
		return;

	if (! me->base) {
		me->base = event_init();
		if (! me->base) {
			fprintf(stderr, "Can't allocate event base\n");
			exit(1);
		}
	}

	/* Listen for notifications from other threads */
	event_set(&me->notify_event, me->notify_receive_fd, \
						EV_READ | EV_PERSIST, thread_libevent_handle, me);
	event_base_set(me->base, &me->notify_event);

	if (event_add(&me->notify_event, 0) == -1) {
		fprintf(stderr, "Can't monitor libevent notify pipe\n");
		exit(1);
	}
	
}

static void worker_thread_create(void * (*handler)(void *), void *arg)
{
	pthread_t       thread;
	pthread_attr_t  attr;
	int             ret;

	pthread_attr_init(&attr);

	if ((ret = pthread_create(&thread, &attr, handler, arg)) != 0) {
			fprintf(stderr, "Can't create thread: %s\n",
								strerror(ret));
			exit(1);
	}
}

static void * thread_loop(void *arg)
{
	assert(arg);
	
	int err;
	sigset_t mask;
	thread_t *thread = arg;
	thread->id = pthread_self();

	sigaddset(&mask, SIGTERM);
	sigaddset(&mask, SIGPIPE);
	sigaddset(&mask, SIGUSR1);
	sigaddset(&mask, SIGUSR2);
	sigaddset(&mask, SIGHUP);
	sigaddset(&mask, SIGINT);
	if(pthread_sigmask(SIG_BLOCK, &mask, NULL) == -1) {
		error_log("sigmask %p failed\n", &mask);
	}

	pthread_mutex_lock(&init_lock);
	init_count++;
	pthread_cond_signal(&init_cond);
	pthread_mutex_unlock(&init_lock);

	event_base_loop(thread->base, 0);
	
	return;
}

static void thread_libevent_handle(int fd, short flag, void *arg)
{
	/* Create new connection. Add callback function to event. */
	assert(arg);

	queue_t *queue;
	char buf = '\0';
	thread_t *thread = arg;
	conn_t *con = NULL;
	accept_t *at = NULL;
	struct list_head *new;

	assert(thread);
	if (read(fd, &buf, 1) != 1)
		debug_log(1, "fd %d Can't read from libevent pipe\n", fd);
	
	/* remove from waiting list, and insert into active list */
	queue = thread->waiting_conns;
	new = queue_head(queue);
	at = list_entry(new, accept_t, list);
	queue_pop(queue);
	if(at == NULL) {
		debug_log(4, "queue is null, count %d\n", queue->count);
		return;
	}
	debug_log(3, "find fd %d, flag %d, data %p, handler %p\n", \
						at->fd, at->flag, at->data, at->handler);
	con = conn_new(at->fd, at->flag, at->handler, NULL, (struct in_addr *)&at->data, \
					thread->base, thread);
	debug_log(3, "client %d, conn %p, active connections %d\n",\
				   	at->fd, con, queue_length(thread->active_conns));

	/* free some unused data */
	mempool_free(at);
}
